package reducers;

import java.io.IOException;
import java.util.HashSet;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
//import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.log4j.Logger;

import types.NodeInfo;

public class ImportNQuadInputFileReducer extends Reducer<Text, NodeInfo, Text, Text> {

private static final Logger sLogger = Logger.getLogger(ImportNQuadInputFileReducer.class);
	
	protected void reduce(Text key, Iterable<NodeInfo> values, Context context) throws IOException, InterruptedException {
	
		// We get all the NodeInfo of a given subject, and write it to disk ready for the next job (SpreadActivator)
				
		String outValue = new String("");	
		
		// Get activation from first node (all of them should have the same activation)
		float activation = 0.0f;	
		//outValue = key.toString();
		
		HashSet<String> objects = new HashSet<String>();
		
		// For each value
		for(NodeInfo val : values) {			
			// Concat connections (destination node and weight)
			String object = val.getNodeAt(0);
			
			// Avoid repetitions
			if(!objects.contains(object)) {			
				outValue += val.getNodeAt(0); // (0): Only one connection
				outValue += " ";
				outValue += val.getWeightAt(0); // (0): Only one connection
				outValue += " ";
				
				objects.add(object);
			}
			
			activation = val.getActivation();
		}
		
		outValue = key + " " + activation + " false " + outValue;
			
		context.write(null, new Text(outValue));
	}
}
